package com.fudan.run.trans;

import java.util.List;
import java.util.Map;

import org.apache.spark.api.java.JavaRDD;

import com.fudan.cfg.trans.FilterFieldsTransDef;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.TransAdapter;

@TransAdapter("filter")
public class FilterFieldsTrans extends TransHandler<FilterFieldsTransDef>{

	@Override
	public void trans(JobRunner jobRunner,FilterFieldsTransDef transNode, VariableRddMap variableMap) {
		JavaRDD<Map<String, Object>> target = variableMap.getRdd(transNode.getTarget());
		JavaRDD<Map<String, Object>> after = target.map(m->{
			List<String> keys = transNode.getExclusion();
			for (String key : keys) {
				m.remove(key);
			}
			return m;
		});
		variableMap.addRdd(transNode.getName(), after);
	}

}
